package org.fight.landowner.remoting.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.fight.landowner.common.DelayExecuteTaskQueue;
import org.fight.landowner.common.serialize.BytesSerializableFactory;
import org.fight.landowner.common.serialize.SerializableType;
import org.fight.landowner.common.util.OSUtil;
import org.fight.landowner.remoting.RemotingServiceStatus;
import org.fight.landowner.remoting.SendCallback;
import org.fight.landowner.remoting.exception.RemotingServiceStatusException;
import org.fight.landowner.remoting.exception.RequestTimeoutException;
import org.fight.landowner.remoting.protocol.RemotingCommand;
import org.fight.landowner.remoting.protocol.RemotingCommandHeader;

/**
 * @Author wangxuehao02
 * @Date 2024/2/1 8:34 PM
 * @Description
 **/
public class NettyRemotingClient extends NettyAbstractRemotingService {
    private String serverIp;
    private int serverPort;

    private Bootstrap bootstrap;
    private Channel channel;
    private EventLoopGroup eventLoopGroup;

    public NettyRemotingClient(String serverIp, int serverPort) {
        this.serverIp = serverIp;
        this.serverPort = serverPort;

        this.eventLoopGroup = OSUtil.isLinux() ? new EpollEventLoopGroup() : new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();
        this.bootstrap
                .group(eventLoopGroup)
                .channel(OSUtil.isLinux() ? EpollSocketChannel.class : NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast(new NettyDecoder())            // 解码器需要保留不完整的网络包，有状态所以不能复用
                                .addLast(NettyEncoder.getInstance())    // 编码器无状态可以复用
                                .addLast(new NettyClientHandler());     // 仅做数据转发可以复用
                    }
                });

        this.responseDelayQueue = new DelayExecuteTaskQueue("NettyRemotingClientResponseDelayQueue", new DelayExecuteTaskQueue.DelayExecuteTaskActuator<ResponseCleanDelayTask>() {
            @Override
            public void execute(ResponseCleanDelayTask delayTask) {
                handlerTimeoutRequest(delayTask.getRequestId());
            }
        });
    }

    public void start() throws InterruptedException {
        if (!status.compareAndSet(RemotingServiceStatus.INIT.getCode(), RemotingServiceStatus.RUNNING.getCode())) {
            throw new RemotingServiceStatusException("NettyRemotingClient status is not init");
        }

        this.channel = this.bootstrap.connect(serverIp, serverPort).sync().channel();

        this.responseDelayQueue.start();
    }

    /**
     * todo 是否需要等待 close
     */
    public void shutdown() throws InterruptedException {
        if (status.compareAndSet(RemotingServiceStatus.INIT.getCode(), RemotingServiceStatus.STOP.getCode())) {
            return;
        }

        if (status.compareAndSet(RemotingServiceStatus.RUNNING.getCode(), RemotingServiceStatus.STOP.getCode())) {
            this.channel.close().sync();
            this.eventLoopGroup.shutdownGracefully();
            this.responseDelayQueue.stop();
        }
    }

    public String getServerIp() {
        return serverIp;
    }

    public int getServerPort() {
        return serverPort;
    }

    public void asyncSend(RemotingCommand remotingCommand, long timeoutMillis, SendCallback sendCallback) {
        this.asyncSend(this.channel, remotingCommand, timeoutMillis, sendCallback);
    }

    public void asyncSend(int code, Object data, SerializableType serializableType, long timeoutMillis, SendCallback sendCallback) {
        RemotingCommand request = new RemotingCommand();
        RemotingCommandHeader header = RemotingCommandHeader.buildRequestRemotingCommandHeader();

        header.setCode((short) code);
        header.setSerializableProtocol(serializableType);

        request.setHeader(header);
        request.setData(BytesSerializableFactory.serialize(data, serializableType));

        asyncSend(request, timeoutMillis, sendCallback);
    }

    public void asyncSend(int code, Object data, long timeoutMillis, SendCallback sendCallback) {
        this.asyncSend(code, data, SerializableType.JSON, timeoutMillis, sendCallback);
    }

    public <T> T syncSend(RemotingCommand remotingCommand, long timeoutMillis, Class<T> responseClazz) throws RequestTimeoutException {
        return this.syncSend(channel, remotingCommand, timeoutMillis, responseClazz);
    }

    public <T> T syncSend(int code, Object data, SerializableType serializableType, long timeoutMillis, Class<T> responseClazz) throws RequestTimeoutException {
        RemotingCommand request = new RemotingCommand();
        RemotingCommandHeader header = RemotingCommandHeader.buildRequestRemotingCommandHeader();

        header.setCode((short) code);
        header.setSerializableProtocol(serializableType);

        request.setHeader(header);
        request.setData(BytesSerializableFactory.serialize(data, serializableType));

        return syncSend(request, timeoutMillis, responseClazz);
    }

    /**
     * 默认使用 JSON 序列化
     * @param code
     * @param data
     * @param timeoutMillis
     * @param <T>
     * @return
     */
    public <T> T syncSend(int code, Object data, long timeoutMillis, Class<T> responseClazz) throws RequestTimeoutException {
        return syncSend(code, data, SerializableType.JSON, timeoutMillis, responseClazz);
    }

    public void noResponseSend(RemotingCommand remotingCommand) {
        this.noResponseSend(channel, remotingCommand);
    }

    public void noResponseSend(int code, Object data, SerializableType serializableType) {
        RemotingCommand request = new RemotingCommand();
        RemotingCommandHeader header = RemotingCommandHeader.buildRequestRemotingCommandHeader();

        header.setCode((short) code);
        header.setSerializableProtocol(serializableType);

        request.setHeader(header);
        request.setData(BytesSerializableFactory.serialize(data, serializableType));

        noResponseSend(request);
    }

    public void noResponseSend(int code, Object data) {
        noResponseSend(code, data, SerializableType.JSON);
    }

    @ChannelHandler.Sharable
    private class NettyClientHandler extends SimpleChannelInboundHandler<RemotingCommand> {

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingCommand remotingCommand) throws Exception {
            processRemotingCommand(channelHandlerContext.channel(), remotingCommand);
        }
    }
}
